package org.nbict.iot.trash;

import com.alibaba.fastjson.JSONObject;
import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

/**
 * Created by songseven on 18/6/19.
 */
@Deprecated
public class Log4MongoBolt extends BaseBasicBolt {

    @Override
    public void execute(Tuple input, BasicOutputCollector
            collector) {

        Long offset = (Long) input.getValueByField("offset");
        String parsed = (String) input.getValueByField("value");
        JSONObject object = JSONObject.parseObject(parsed);
        JSONObject body = (JSONObject) object.getJSONObject("car_packet_body");
        JSONObject header = (JSONObject) object.getJSONObject("car_packet_header");

        String platform = (String) input.getValueByField("topic");
        String vin = header.getString("car_vin");
        int cmd = header.getInteger("header_cmd");
        Long timestamp = body.getLong("timestamp");

        collector.emit(new Values(platform, vin, cmd, timestamp, body));
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("platform_id", "car_vin","action", "timestamp", "info"));
    }


//    @Override
//    public void execute(Tuple input, BasicOutputCollector
//            collector) {
//        String platform = (String) input.getValueByField("topic");
//        Long offset = (Long) input.getValueByField("offset");
//
//        System.out.println(offset);
//
//        String parsed = (String) input.getValueByField("value");
//        JSONObject object = JSONObject.parseObject(parsed);
//        JSONObject body = (JSONObject) object.getJSONObject("car_packet_body");
//        JSONObject header = (JSONObject) object.getJSONObject("car_packet_header");
//
//        Long timestamp = body.getLong("timestamp"); //时间戳
//        int cmd = header.getInteger("header_cmd");
//        String vin = header.getString("car_vin");
//
//        //实时信息上报或者补发信息上报
//        Map history = new HashMap();
//        history.put("platform_id", platform);
//        history.put("car_vin", vin);
//        history.put("action", cmd);
//        history.put("timestamp", timestamp);
//        history.put("info", body.toJSONString());
//        repo.insert("car_history", new BasicDBObject(history));
//    }

//    @Override
//    public void execute(Tuple input) {
//        String platform = (String) input.getValueByField("topic");
//        Long offset = (Long) input.getValueByField("offset");
//
//        System.out.println(offset);
//
//        String parsed = (String) input.getValueByField("value");
//        JSONObject object = JSONObject.parseObject(parsed);
//        JSONObject body = (JSONObject) object.getJSONObject("car_packet_body");
//        JSONObject header = (JSONObject) object.getJSONObject("car_packet_header");
//
//        Long timestamp = body.getLong("timestamp"); //时间戳
//        int cmd = header.getInteger("header_cmd");
//        String vin = header.getString("car_vin");
//
//        //实时信息上报或者补发信息上报
//        Map history = new HashMap();
//        history.put("platform_id", platform);
//        history.put("car_vin", vin);
//        history.put("action", cmd);
//        history.put("timestamp", timestamp);
//        history.put("info", body.toJSONString()); //debug
//
////        try {
////            if (batchHelper.shouldHandle(input)) {
////                batchHelper.addBatch(input);
////            }
////
////            if (batchHelper.shouldFlush()) {
////                flushTuples();
////                batchHelper.ack();
////            }
////        } catch (Exception e) {
////            batchHelper.fail(e);
////        }
//
////        repo.insert("car_history", new BasicDBObject(history));
//
//        _collector.ack(input);
//    }
//    private void flushTuples() {
//
//    }

}
